package com.shujia.flink.tf;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Demo1Map {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //读取数据
        DataStream<String> studentDS = env.readTextFile("flink/data/students.csv");

        /*
         * 1、使用接口的实现类
         */
        DataStream<String> mapDS = studentDS.map(new MapFun());

        //mapDS.print();


        /*
         * 2、使用匿名内部类
         */

        DataStream<Integer> ageDS = studentDS
                .map(new MapFunction<String, Integer>() {
                    @Override
                    public Integer map(String value) throws Exception {
                        String age = value.split(",")[2];
                        return Integer.parseInt(age);
                    }
                });

        //ageDS.print();

        /*
         * 3、lambda表达式
         * 如果返回值是嵌套结构需要指定返回值的类型信息
         */

        DataStream<Tuple2<String, Integer>> sexDS = studentDS
                .map(value -> Tuple2.of(value.split(",")[3], 1), Types.TUPLE(Types.STRING, Types.INT));

        sexDS.print();


        env.execute();

    }
}

class MapFun implements MapFunction<String, String> {
    @Override
    public String map(String student) throws Exception {
        return student.split(",")[4];
    }
}
